# -*- coding: utf-8 -*-
#
# Copyright 2007 - Celso Pinto <cpinto@7syntax.com>
#
# Licensed under the Apache License, Version 2.0 (the "License"); 
# you may not use this file except in compliance with the License. 
# You may obtain a copy of the License at 
#
#      http://www.apache.org/licenses/LICENSE-2.0 
#
# Unless required by applicable law or agreed to in writing, software 
# distributed under the License is distributed on an "AS IS" BASIS, 
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
# See the License for the specific language governing permissions and 
# limitations under the License.
#

import stomp
import Queue

from processors import SerializedProcessor, ConcurrentProcessor
from threading import Thread

__all__ = ['QUEUE','TOPIC','Config','Context','Message','ServerConnection']

(
    QUEUE,
    TOPIC,
) = xrange(2)

DEST_PREFIX = {QUEUE:'/queue/%s',TOPIC:'/topic/%s'}

class Config(object):
    def __init__(self,destination, settings):
        self.config =   {
                            'type': QUEUE,
                            'auto_acknowledge': True,
                            'transactional': False,
                            'concurrent': False,
                            'destination': destination,
                            'custom_headers' : {},
                            'min_pool_size':1,
                            'max_pool_size':1,
                        }

        self.config.update(settings)
        self.__validate_config()

    def __validate_config(self):
        # type validity
        if not self.type in (QUEUE,TOPIC,):
            raise InvalidSettings('Invalid type')

        # if concurrent, must have min_pool_size and max_pool_size > 0
        if self.concurrent:
            if min(self.min_pool_size,self.max_pool_size) < 1:
                raise InvalidSettings('Pool size must be >= 1')
            if self.min_pool_size > self.max_pool_size:
                raise InvalidSettings('max pool size smaller than min pool size')

        if not callable(self.worker):
            raise InvalidSettings('Worker not callable')
        

    def __getattr__(self,name):
        if not name in self.config:
            return super(Config,self).__getattribute__(name)
        if name == 'destination':
            return DEST_PREFIX[self.config['type']]  % self.config['destination']

        return self.config[name]

class Context(object):
    def __init__(self,transaction=False):
        self.transaction = transaction
        self.rollback = False
        self.transaction_id = None

    def __get_transaction(self): return self.transaction
    in_transaction = property(__get_transaction)

    def set_rollback(self): self.rollback = True

class MessageConsumer(Thread):
    def __init__(self,config):
        super(MessageConsumer,self).__init__()
        self.config = config
        self.q = Queue.Queue()
        self.die = False
        if config.concurrent:
            self.processor = ConcurrentProcessor(config.min_pool_size,config.max_pool_size)
        else:
            self.processor = SerializedProcessor()

    def on_message(self,m):
        self.q.put_nowait(m)

    def stop(self):
        self.die = True
        self.processor.stop()

    def run(self):
        while(not self.die):
            try:
                m = self.q.get(timeout=5)
                self.processor.process(m)
            except: #Queue.Empty
                pass

class Message(object):
    def __init__(self,headers,body,conn,config,ctx):
        self.headers = headers
        self.body = body
        self.conn = conn
        self.config = config
        self.context = ctx

    def acknowledge(self):
        if not self.config.auto_acknowledge:
            headers = dict(message_id=self.headers['message-id'])
            if self.context.in_transaction:
                headers['transaction'] = self.context.transaction_id

            self.conn.ack(headers)

    def rollback(self):
        if self.context.in_transaction:
            self.context.set_rollback()

class Session(stomp.ConnectionListener):
    def __init__(self,servers,consumers):
        self.servers = servers
        self.consumers = dict()
        for dest,conf in consumers.iteritems():
            c = Config(dest,conf)
            self.consumers[c.destination] = MessageConsumer(c)

        self.conn = stomp.Connection(self.servers)
        self.conn.add_listener(self)

    def on_connecting(self,host_post):
        self.conn.connect()

    def on_connected(self,headers,body):
        self.subscribe_all()

    def on_message(self,headers,body):
        consumer = self.consumers[headers['destination']]
        ctx = Context(consumer.config.transactional)

        m = Message(headers,
                    body,
                    self.conn,
                    consumer.config,
                    ctx)

        consumer.on_message(m)

    def subscribe_all(self):
        for c in self.consumers.values():
            if c.config.auto_acknowledge:
                ack = 'auto'
            else:
                ack = 'client'

            try:
                headers = c.config.custom_headers.get('subscribe',{})
            except:
                headers = {}
                
            self.conn.subscribe(headers,destination=c.config.destination,ack=ack)

    def unsubscribe_all(self):
        for c in self.consumers.values():
            self.conn.unsubscribe(destination=c.config.destination)

    def start(self):
        for c in self.consumers.values():
            c.start()

        self.conn.start()

    def stop(self):
        try:
            # first unsubscribe so that no new messages are received
            # this allows us to start a separate process with new code, etc
            # while allowing all work in progress to finish
            self.unsubscribe_all()
        except:
            pass

        # then signal handler threads to stop
        consumers = self.consumers.values()
        [ c.stop() for c in consumers ]

        # and wait for them to stop doing whatever they're doing
        [ c.join() for c in consumers ]

        # disconnect from server
        self.conn.disconnect()
        #self.conn.stop()
        
